package com.asiainfo.dacp.datastash.broker.flow.core.service;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.enums.ErrorCodeEnum;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStepTypeEnum;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitStep;
import com.asiainfo.dacp.datastash.broker.core.vo.StashUnitTask;
import com.asiainfo.dacp.datastash.broker.flow.core.domain.Flow;
import com.asiainfo.dacp.datastash.broker.flow.sharding.service.ShardingService;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnit;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnitStep;
import com.google.common.base.Preconditions;
import com.google.common.base.Strings;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;

@Service
public class FlowCompileService {
	private static Logger LOG = LoggerFactory.getLogger(FlowCompileService.class);
	@Autowired
	private ShardingService shardingService;

	public void compile(Flow flow) throws BrokerException {

		List<StashUnitTask> tasks = buildTask(flow);
		if (tasks.size() == 0)
			throw new BrokerException(ErrorCodeEnum.CONFIG_ERROR, "编译任务数为0，检查流程配置或分片配置是否正确.");

		HashMap<String, StashUnitTask> map = Maps.newHashMap();
		for (StashUnitTask task : tasks) {
			map.put(task.getTaskProcId(), task);
		}
		flow.setTasks(map);
		flow.setTaskCnt(tasks.size());
	}

	public List<StashUnitTask> buildTask(Flow flow) throws BrokerException {
		// build-task-link
		// 限定： 不允许分支出现
		Map<String, StashUnitTask> tasks = Maps.newConcurrentMap();

		DataStashUnit unit = flow.getUnit();

		List<DataStashUnitStep> rootSteps = unit.getRootSteps();
		StashUnitTask rootTask = null;
		if (rootSteps != null && rootSteps.size() > 0) {
			DataStashUnitStep step = rootSteps.get(0);
			StashUnitTask preTask = null;
			do {
				String stepType = step.getStepDefine().getStepType();

				Preconditions.checkNotNull(stepType);
				if (stepType.equals(UnitStepTypeEnum.Datax.value()) || stepType.equals(UnitStepTypeEnum.DP.value())
						|| stepType.equals(UnitStepTypeEnum.Sharding.value())) {

					if (preTask == null) {
						preTask = new StashUnitTask(stepType, flow);
						if (rootTask == null) {
							rootTask = preTask;
						}
					} else {
						if (!preTask.getTaskType().equals(stepType)) {
							StashUnitTask task = new StashUnitTask(stepType, flow);
							task.addPreTask(preTask);
							preTask.addNextTask(task);
							preTask = task;
						}
					}
					if (!tasks.containsKey(preTask.getTaskProcId())) {
						tasks.put(preTask.getTaskProcId(), preTask);
					}
					preTask.addTaskStep(buildStashUnitStep(step, preTask));
				}

				List<DataStashUnitStep> nextSteps = step.getNextSteps();
				step = null;
				if (nextSteps != null && nextSteps.size() > 0) {
					step = nextSteps.get(0);
				}
			} while (step != null);
		}

		// fill-link
		// sharding 替换处理
		Preconditions.checkNotNull(rootTask);
		StashUnitTask curTask = rootTask;

		StashUnitTask preTask = null;
		StashUnitTask nextTask = null;
		do {
			if (!Strings.isNullOrEmpty(curTask.nextProcId())) {
				nextTask = tasks.get(curTask.nextProcId());
			} else {
				nextTask = null;
			}

			if (curTask.getTaskType().equals(UnitStepTypeEnum.Sharding.value())) {
				StashUnitStep step = curTask.getTaskSteps().get(0);
				DataStashUnitStep stepByInx = unit.getStepIndex().get(step.getStepIdx());
				// 清理掉前后task对当前task索引
				if (preTask != null) {
					preTask.cleanNextTask();
				}
				if (nextTask != null) {
					nextTask.cleanPreTask();
				}
				buildShardingTask(stepByInx, flow, preTask, nextTask, tasks);
				tasks.remove(curTask.getTaskProcId());
			}

			preTask = curTask;
			curTask = nextTask;
		} while (curTask != null);

		return Lists.newArrayList(tasks.values());
	}

	public void buildShardingTask(DataStashUnitStep stepByInx, Flow flow, StashUnitTask pre, StashUnitTask next,
			Map<String, StashUnitTask> tasks) throws BrokerException {

		List<List<Map<String, String>>> shardContents = shardingService.sharding(stepByInx, flow);

		for (int i = 0; i < shardContents.size(); i++) {

			String taskType = stepByInx.getChildren().get(0).getStepDefine().getStepType();

			StashUnitTask shardTask = new StashUnitTask(taskType, flow);
			shardTask.setTaskContent(shardContents.get(i));

			tasks.put(shardTask.getTaskProcId(), shardTask);
			for (DataStashUnitStep cStep : stepByInx.getChildren()) {
				StashUnitStep stashUnitStep = buildStashUnitStep(cStep, shardTask);
				shardTask.addTaskStep(stashUnitStep);
			}

			if (pre != null) {
				pre.addNextTask(shardTask);
				shardTask.addPreTask(pre);
			}
			if (next != null) {
				next.addPreTask(shardTask);
				shardTask.addNextTask(next);
			}
		}
	}

	private StashUnitStep buildStashUnitStep(DataStashUnitStep stepDomain, StashUnitTask curTask) {

		StashUnitStep stashUnitStep = new StashUnitStep();
		stashUnitStep.setStepIdx(stepDomain.getStepIndex());

		if (curTask != null) {
			stashUnitStep.setTaskProcId(curTask.getTaskProcId());
			stashUnitStep.setUnitCode(curTask.getUnitCode());
		}

		stashUnitStep.setNextStepIdxs(stepDomain.getNextStepIdxs());
		stashUnitStep.setPreStepIdxs(stepDomain.getPreStepIdxs());
		stashUnitStep.setParentStepIdx(stepDomain.getParentStepIdx());
		stashUnitStep.setIgnoreErrStepIdxs(stepDomain.getIgnoreErrStepIdxs());
		stashUnitStep.setStepCode(stepDomain.getStepInst());
		stashUnitStep.setStepCfgVal(stepDomain.getStepCfgVal());

		return stashUnitStep;
	}

}
